/*
 * Author: 李波(Edwin Lee)
 * Date: 2024-08-07 04:31:23
 * FilePath: /mprpc/src/rpcprovider.cc
 * LastEditTime: 2024-09-03 09:13:00
 * version: 1.0
 * Description: 
 */
#include "rpcprovider.h"
#include "mprpcapplication.h"
#include "rpcheader.pb.h"
#include "logger.h"
#include "zookeeperutil.h"

// 可以发布rpc方法的函数接口(在RpcProvider注册一个rpc对象，存储到serviceMap_)
void RpcProvider::NotifyService(google::protobuf::Service* service)
{
    ServiceInfo serviceInfo;

    // 获取服务对象的描述信息（protobuf自带函数）
    const google::protobuf::ServiceDescriptor* pserviceDesc = service->GetDescriptor();
    // 获取服务的名字
    std::string serviceName = pserviceDesc->name();
    // 获取服务对象service的方法的数量
    int methodCnt = pserviceDesc->method_count();

    LOG_INFO("serviceName:%s",serviceName);
    for(int i=0; i < methodCnt; ++i)
    {
        // 获取了服务对象指定下标的服务方法的描述（抽象描述）
        const google::protobuf::MethodDescriptor* pmethodDesc = pserviceDesc->method(i);
        std::string methodName = pmethodDesc->name();
        serviceInfo.methodMap_.insert({methodName,pmethodDesc});

        //std::cout << "methodName: " << methodName <<std::endl;
        LOG_INFO("methodName:%s",methodName);
    }
    serviceInfo.service_ = service;
    serviceMap_.insert({serviceName,serviceInfo});
}

// 启动rpc服务节点，开始提供rpc远程过程调用服务
void RpcProvider::Run()
{
    MprpcConfig config = MprpcApplication::getInstance().getConfig();
    std::string ip = config.load("rpcserverip");
    uint16_t port = atoi(config.load("rpcserverport").c_str());
    int muduoThreadNum = atoi(config.load("muduothreadnum").c_str());
    muduo::net::InetAddress address(ip,port);

    // 创建TcpServer对象
    muduo::net::TcpServer server(&eventloop_,address,"RpcProvider");
    // 绑定连接回调和消息读写回调方法，分离了网络代码和业务代码
    server.setConnectionCallback(std::bind(&RpcProvider::onConnection,this,std::placeholders::_1));
    // 设置muduo库的线程数量
    server.setMessageCallback(std::bind(&RpcProvider::onMessage,this,std::placeholders::_1,
        std::placeholders::_2,std::placeholders::_3));
    // 设置muduo库的线程数量
    server.setThreadNum(muduoThreadNum);

    // 把当前rpc节点上要发布的服务全部注册到zk上面，让rpc client可以从zk上发现服务
    // session timeout 30s   zk会在1/3timeout时间发送ping消息作为心跳
    LOG_INFO("zk connect begin!");
    ZkClient zkcli;
    // 连接zk
    zkcli.start();
    // service_name为永久性节点 method_name为临时节点
    for(auto& sp:serviceMap_)
    {
        // /service_name
        std::string service_path = "/"+sp.first;
        zkcli.create(service_path.c_str(),nullptr,0);
        for(auto& mp:sp.second.methodMap_)
        {
            // /service_name/method_name
            std::string method_path = service_path + "/" + mp.first;
            char method_path_data[128]{0};
            sprintf(method_path_data,"%s:%d",ip.c_str(),port);
            // ZOO_EPHEMERAL 表示此节点为临时节点
            zkcli.create(method_path.c_str(),method_path_data,strlen(method_path_data),ZOO_EPHEMERAL);
        }
    }
    LOG_INFO("RpcProvider start service at ip:%s port:%d",ip.c_str(),port);
    // 启动网络服务
    server.start();
    eventloop_.loop();

}

void RpcProvider::onConnection(const muduo::net::TcpConnectionPtr& conn)
{
    if(!conn->connected())
    {
        conn->shutdown();
    }
}
/*
    通信协议：
    header_size(4个字节，header_str的长度) + header_str(rpcHeader的内容，包含了args_str的长度) +args_str
*/

void RpcProvider::onMessage(const muduo::net::TcpConnectionPtr& conn, 
    muduo::net::Buffer* buffer, muduo::Timestamp time)
{
    // 网络上接收的远程rpc调用请求的字符流
    std::string recvBuf = buffer->retrieveAllAsString();

    // 从字符流中读取前4个字节的内容
    uint32_t header_size = 0;
    recvBuf.copy((char*)&header_size,4);

    // 根据header_size读取数据头的原始字符流,反序列化数据，得到rpc请求的详细信息
    std::string rpc_header_str = recvBuf.substr(4,header_size);
    mprpc:: RpcHeader rpcHeader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size;
    if(rpcHeader.ParseFromString(rpc_header_str))
    {
        // 数据头反序列化成功
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else 
    {
        // 数据头反序列化失败
        std::cout << "rpc_header_str: " << rpc_header_str <<"parse error!" << std::endl;
        return;
    }
    // 获取rpc方法参数的字符流数据
    std::string args_str = recvBuf.substr(4+header_size,args_size);

    // 打印调试信息
    std::cout <<"================================================" << std::endl;
    std::cout << "header_size: " << header_size << std::endl;
    std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    std::cout << "service_name: " << service_name << std::endl;
    std::cout << "method_name: " << method_name << std::endl;
    std::cout << "args_str: " << args_str << std::endl;
    std::cout <<"================================================" << std::endl;

    // 获取service对象和method对象
    auto it = serviceMap_.find(service_name);
    if(it == serviceMap_.end())
    {
        std::cout << service_name << "is not exiit!" << std::endl;
        return;
    }
    // 获取service对象，new UserService
    google::protobuf::Service* service = it->second.service_;

    auto mit = it->second.methodMap_.find(method_name);
    if(mit == it->second.methodMap_.end())
    {
        std::cout << service_name << ": " << method_name << "is not exit!" << std::endl;
        return;
    }
    // 获取method对象 login
    const google::protobuf::MethodDescriptor* method = mit->second;

    // 生成rpc方法调用的请求reques和相应response的数据
    google::protobuf::Message* request = service->GetRequestPrototype(method).New();
    if(!request->ParseFromString(args_str))
    {
        std::cout << "request pase error,content: " << args_str << std::endl;
        return;
    }
    google::protobuf::Message* response = service->GetResponsePrototype(method).New();

    // 给下面的method方法的调用，绑定一个Closure的回调函数
    google::protobuf::Closure* done = 
        google::protobuf::NewCallback<RpcProvider,const muduo::net::TcpConnectionPtr&,
        google::protobuf::Message*>(this,&RpcProvider::SendRpcResponse,conn,response);
    // 在框架上根据远端rpc请求，调用当前rpc节点上发布的方法
    // new UserService().Login(controller,request,response,done)
    service->CallMethod(method,nullptr,request,response,done);

}

void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr& conn,google::protobuf::Message* response)
{
    std::string response_str;
    if(response->SerializeToString(&response_str))
    {
        // 序列化成功后，通过网络把rpc方法执行的结果发送回rpc的调用方
        conn->send(response_str);
        // 模拟http短链接服务，由rpcprovider主动断开连接
        conn->shutdown();
    }
    else
    {
        std::cout << "serialize response_str error!" << std::endl;
    }
}